using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{
	public class ThriftColumnFamilyTemplate<K,N> : ColumnFamilyTemplate<K,N>
	{
        public ThriftColumnFamilyTemplate(Keyspace keyspace, String columnFamily,
            Serializer<K> keySerializer, Serializer<N> topSerializer) :
            base(keyspace, columnFamily, keySerializer, topSerializer)
        {
        }
        public override T DoExecuteSlice<T>(K key, HSlicePredicate<N> predicate, ColumnFamilyRowMapper<K, N, T> mapper)
        {
            return mapper.MapRow(DoExecuteSlice(key,predicate));
        }
        public override ColumnFamilyResult<K, N> DoExecuteSlice(K key, HSlicePredicate<N> workingSlicePredicate)
        {
            return new ColumnFamilyResultWrapper<K, N>(KeySerializer, TopSerializer, 
                SliceInternal(key, workingSlicePredicate));
        }
        public override ColumnFamilyResult<K, N> DoExecuteMultigetSlice(IEnumerable<K> keys, HSlicePredicate<N> workingSlicePredicate)
        {
            return new ColumnFamilyResultWrapper<K, N>(KeySerializer, TopSerializer, 
                MultigetSliceInternal(keys, workingSlicePredicate));
        }
        public override MappedColumnFamilyResult<K, N, V> DoExecuteMultigetSlice<V>(IEnumerable<K> keys, HSlicePredicate<N> workingSlicePredicate, ColumnFamilyRowMapper<K, N, V> mapper)
        {
            return new MappedColumnFamilyResultWrapper<K,N,V>(KeySerializer, TopSerializer, 
                MultigetSliceInternal(keys, workingSlicePredicate), mapper);
        }
        public override ColumnFamilyResult<K, N> DoExecuteIndexedSlices<V>(IndexedSlicesPredicate<K, N, V> predicate) 
        {
            return new ColumnFamilyResultWrapper<K, N>(KeySerializer, TopSerializer,
                IndexedSlicesInternal(predicate, ActiveSlicePredicate));
        }
        public override ColumnFamilyResult<K, N> DoExecuteIndexedSlices<V>(IndexedSlicesPredicate<K, N, V> predicate,
            HSlicePredicate<N> slicePredicate)
        {
            return new ColumnFamilyResultWrapper<K, N>(KeySerializer, TopSerializer,
                IndexedSlicesInternal(predicate, slicePredicate));
        }
        public override MappedColumnFamilyResult<K, N, R> DoExecuteIndexedSlices<R, V>(IndexedSlicesPredicate<K, N, V> predicate, ColumnFamilyRowMapper<K, N, R> mapper)
        {
            return new MappedColumnFamilyResultWrapper<K,N,R>(KeySerializer, TopSerializer, 
                IndexedSlicesInternal(predicate, ActiveSlicePredicate), mapper);
        }
        public override MappedColumnFamilyResult<K, N, R> DoExecuteIndexedSlices<R, V>(IndexedSlicesPredicate<K, N, V> predicate, HSlicePredicate<N> slicePredicate, ColumnFamilyRowMapper<K, N, R> mapper)
        {
            return new MappedColumnFamilyResultWrapper<K,N,R>(KeySerializer, TopSerializer, 
                IndexedSlicesInternal(predicate, slicePredicate), mapper);
        }
        private ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> SliceInternal(K key,
            HSlicePredicate<N> workingSlicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<ColumnOrSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[],List<ColumnOrSuperColumn>> cosc = new Dictionary<byte[], List<ColumnOrSuperColumn>>();
                    byte[] sKey = KeySerializer.ToBytes(key);
                    cosc[sKey] =  cassandra.get_slice(sKey, ColumnParent, workingSlicePredicate.ToThrift(), 
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ, ColumnFamily)));
                    return cosc;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<ColumnOrSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<ColumnOrSuperColumn>>>(x);
            op.OperationType = OperationType.READ;
            return ((ExecutingKeyspace)Keyspace).DoExecuteOperation(op);
        }
        private ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> MultigetSliceInternal(IEnumerable<K> keys,
            HSlicePredicate<N> workingSlicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<ColumnOrSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<K> keyList = new List<K>();
                    IEnumerator<K> enumerator = keys.GetEnumerator();
                    while (enumerator.MoveNext())
                    {
                        keyList.Add(enumerator.Current);
                    }
                    List<N> nameList = new List<N>(ColumnValueSerializers.Keys);
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cosc = cassandra.multiget_slice(
                        KeySerializer.ToBytesList(keyList), ColumnParent,
                        (workingSlicePredicate == null ? ActiveSlicePredicate.SetColumnNames(nameList).ToThrift() : workingSlicePredicate.ToThrift()),
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ, ColumnFamily)));
                    return cosc;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<ColumnOrSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<ColumnOrSuperColumn>>>(x);
            op.OperationType = OperationType.READ;
            return ((ExecutingKeyspace)Keyspace).DoExecuteOperation(op);
        }
        private ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> IndexedSlicesInternal<V>(IndexedSlicesPredicate<K, N, V> predicate, 
            HSlicePredicate<N> workingSlicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<ColumnOrSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    
                    List<KeySlice> keySlices = cassandra.get_indexed_slices(ColumnParent, predicate.ToThrift(), 
                        workingSlicePredicate.ToThrift(),
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ, ColumnFamily)));
                    if (keySlices == null || keySlices.Count == 0)
                    {
                        return new Dictionary<byte[], List<ColumnOrSuperColumn>>(0);
                    }
                    Dictionary<byte[], List<ColumnOrSuperColumn>> ret = 
                        new Dictionary<byte[], List<ColumnOrSuperColumn>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = keySlice.Columns;
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<ColumnOrSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<ColumnOrSuperColumn>>>(x);
            op.OperationType = OperationType.READ;
            return ((ExecutingKeyspace)Keyspace).DoExecuteOperation(op);
        }
	}
}